package com.im.consumer.coder;

import com.im.consumer.message.Message;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;

public class MessageSerializer implements Serializer<Message> {

    @Override
    public byte[] serialize(String topic, Message data) {
        if (data == null) {
            return null;
        }

        try {
            // 获取消息头和消息体的字节数组
            byte[] headerBytes = data.getHeader().toByteArray();
            byte[] bodyBytes = data.getBody().toByteArray();

            // 计算消息的总长度
            int totalLength = 4 + 4 + 4 + 4 + headerBytes.length + bodyBytes.length;
            ByteBuffer buffer = ByteBuffer.allocate(totalLength);
            buffer.putInt(data.getPacketBoundary()); // packetBoundary
            buffer.putInt(data.getVersion());        // version
            buffer.putInt(headerBytes.length);       // headerLength
            buffer.putInt(bodyBytes.length);         // bodyLength
            buffer.put(headerBytes);                 // header
            buffer.put(bodyBytes);                   // body

            return buffer.array();
        } catch (Exception e) {
            throw new RuntimeException("Error while serializing Message", e);
        }
    }
}
